package cn.mxleader.kafka.mxleaderspringkafkademo;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.http.MediaType;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

import java.awt.*;

@SpringBootApplication
@RestController
public class MxleaderSpringKafkaDemoApplication {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;


    @PostMapping(value = "/produce/{flag}",
            consumes = MediaType.TEXT_PLAIN_VALUE,
            produces = MediaType.APPLICATION_JSON_VALUE)
    public ApiResponse<String> produceKafkaMessage(@PathVariable Integer flag,
                                                   @RequestBody String message) {
        String msg = message;
        if (flag <= 0) {
            msg = "Default Hello Kafka Message 默认消息 - Hello Kafka";
        }
        kafkaTemplate.send("source_topic", msg);
        return new ApiResponse<>("Invoke Method: produceKafkaMessage()",
                ApiResponse.CODE.SUCCESS,
                "消息内容: '[" + msg + "]' 成功发送到Kafka服务");
    }
/*

    @KafkaListener(topics = "source_topic", id = "source_handler")
    public void kafkaListener1(String message) {
        System.out.println("Message received from Producer to Consumer");
        System.out.println(message);
    }
*/

    public static void main(String[] args) {
        SpringApplication.run(MxleaderSpringKafkaDemoApplication.class, args);
    }
}
